package fetcher

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"regexp"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"ug-uploads/config"
	"ug-uploads/db"
	"ug-uploads/initialize"
	"ug-uploads/logx"
	"ug-uploads/uploader"

	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

type VideoListResponse struct {
	Data struct {
		List []struct {
			ID        string  `json:"id"`
			URL       string  `json:"url"`
			PostUrl   string  `json:"poster_url"`
			Filename  string  `json:"filename"`
			Format    string  `json:"format"`
			Size      int64   `json:"size"`
			Duration  float64 `json:"duration"`
			Signature string  `json:"signature"`
		} `json:"list"`
	} `json:"data"`
}

type MaterialPkg struct {
	ID             int64     `json:"id"`
	Name           string    `json:"name"`
	UserId         int64     `json:"userId"`
	IsPublic       int64     `json:"isPublic"`
	AutoUnlockTime time.Time `json:"autoUnlockTime"`
}

type Material struct {
	Name      string  `json:"name"`       // 素材名称
	File      string  `json:"file"`       // 视频文件名
	Size      int64   `json:"size"`       // 文件大小（字节）
	Dimension string  `json:"dimension"`  // 分辨率或占位字段
	Duration  float64 `json:"duration"`   // 时长（秒）
	UserID    int64   `json:"user_id"`    // 用户ID
	PackageID int64   `json:"package_id"` // 所属素材包ID
	Poster    string  `json:"poster"`     // 封面图文件名
	Signature string
}

type CreatePkgResponse struct {
	Code int             `json:"code"`
	Msg  string          `json:"msg"`
	Data json.RawMessage `json:"data"`
}

func FetchAndUpload(ctx context.Context, advertiserID string) error {
	createPkgURL := config.Cfg.CreatePkgURL
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, createPkgURL, nil)

	if err != nil {
		logx.Logger.Error("创建素材包请求失败", zap.Error(err))
		return fmt.Errorf("创建素材包请求失败: %w", err)
	}

	req.Header.Set("x-service-key", "gFnGn~58Hk^dF(+uEtc+cish")
	resp, err := http.DefaultClient.Do(req)

	if err != nil {
		logx.Logger.Error("调用创建素材包接口失败", zap.Error(err))
		return fmt.Errorf("调用创建素材包接口失败: %w", err)
	}

	defer resp.Body.Close()
	bodyBytes, _ := io.ReadAll(resp.Body)

	if resp.StatusCode != http.StatusOK {
		logx.Logger.Error("创建素材包接口响应错误",
			zap.Int("status", resp.StatusCode),
			zap.String("body", string(bodyBytes)))
		return fmt.Errorf("创建素材包接口响应错误: %s", string(bodyBytes))
	}

	var pkgResp CreatePkgResponse

	if err := json.Unmarshal(bodyBytes, &pkgResp); err != nil {
		logx.Logger.Error("解析创建素材包响应失败", zap.Error(err))
		return fmt.Errorf("解析创建素材包响应失败: %w", err)
	}

	logx.Logger.Info("素材包创建成功",
		zap.Int("code", pkgResp.Code),
		zap.String("msg", pkgResp.Msg),
		zap.ByteString("data", pkgResp.Data))

	var pkgData MaterialPkg

	if err := json.Unmarshal(pkgResp.Data, &pkgData); err != nil {
		logx.Logger.Error("解析素材包信息失败", zap.Error(err))
		return fmt.Errorf("解析素材包信息失败: %w", err)
	}

	logx.Logger.Info("素材包信息",
		zap.Int64("pkg_id", pkgData.ID),
		zap.String("pkg_name", pkgData.Name))

	start := time.Now()
	videoChan := make(chan VideoListResponse)
	batch := 10
	materialChan := make(chan Material, batch)

	//请求素材资源协程（生产者1）
	go func() {
		defer close(videoChan)
		page := 1
		pageSize := batch
		for {
			select {
			case <-ctx.Done():
				return
			default:
				url := fmt.Sprintf("%s?advertiser_id=%s&page=%d&page_size=%d",
					config.Cfg.BaseURL, advertiserID, page, pageSize)
				req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
				req.Header.Set("Access-Token", config.GetAccessToken())
				resp, err := http.DefaultClient.Do(req)

				if err != nil {
					logx.Logger.Error("请求巨量素材接口失败", zap.String("url", url), zap.Int("page", page))
					continue
				}

				if resp.StatusCode != http.StatusOK {
					resp.Body.Close()
					logx.Logger.Error("请求巨量素材接口失败", zap.Int("status code", resp.StatusCode), zap.String("url", url), zap.Int("page", page))
					continue
				}

				body, err := io.ReadAll(resp.Body)

				if err != nil {
					resp.Body.Close()
					logx.Logger.Error("读取巨量素材接口响应失败", zap.String("url", url), zap.Int("page", page))
					continue
				}

				resp.Body.Close()

				var result VideoListResponse

				if err := json.Unmarshal(body, &result); err != nil {
					logx.Logger.Error("解析巨量素材失败", zap.String("url", url), zap.Int("page", page))
					continue
				}
				// fmt.Println("count", len(result.Data.List))
				// var responseData map[string]any
				// json.Unmarshal(body, &responseData)
				// fmt.Println("body", responseData)
				videoChan <- result
				list := result.Data.List

				if len(list) == 0 {
					jsonStr, err := json.Marshal(result)
					if err != nil {
						logx.Logger.Info("巨量素材接口分页数据为空，结束请求",
							zap.String("advertiser_id", advertiserID),
							zap.Int("page", page),
							zap.String("result", ""),
							zap.Error(err))
						return
					}
					logx.Logger.Info("巨量素材接口分页数据为空，结束请求",
						zap.String("advertiser_id", advertiserID),
						zap.Int("page", page),
						zap.String("result", string(jsonStr)))
					return
				}

				logx.Logger.Info("巨量素材接口分页请求成功",
					zap.String("advertiser_id", advertiserID),
					zap.Int("page", page),
					zap.Int("条数", len(list)))

				if len(list) < pageSize {
					logx.Logger.Info("巨量素材接口已到最后一页",
						zap.String("advertiser_id", advertiserID),
						zap.Int("page", page))
					return
				}

				page++
				time.Sleep(500 * time.Millisecond)
			}
		}
	}()

	//素材上传（生产者2）
	go func() {
		defer close(materialChan)
		for {
			select {
			case <-ctx.Done():
				return
			case result, ok := <-videoChan: //一次处理一个请求列表的数据即可，下方g.Go再根据视频数量开辟协程
				if !ok {
					logx.Logger.Info("巨量素材请求接口通道已关闭")
					//无缓冲的管道，直接返回即可
					return
				}

				videoList := result.Data.List

				if len(videoList) == 0 {
					logx.Logger.Info("待上传素材管道无视频数据", zap.String("advertiser_id", advertiserID))
					return
				}

				re := regexp.MustCompile(`^(.+?)-([^/-]+)$`)

				maxConcurrent := min(len(videoList), runtime.NumCPU()*4)
				logx.Logger.Info("开始上传素材", zap.Int("并发数", maxConcurrent))
				g, gctx := errgroup.WithContext(ctx) //子协程子协程子协程
				g.SetLimit(maxConcurrent)

				for _, value := range videoList {
					v := value
					g.Go(func() error {
						if v.Format != "MP4" && v.Format != "mp4" {
							logx.Logger.Info("非MP4文件, 跳过", zap.String("format", v.Format))
							return nil
						}

						if !re.MatchString(v.Filename) {
							logx.Logger.Info("文件名不符", zap.String("filename", v.Filename))
							return nil
						}

						exists, err := db.AlreadyUploaded(v.Signature)

						if err != nil || exists {
							if err != nil {
								logx.Logger.Error("检查上传重复失败: ", zap.Error(err))
								return fmt.Errorf("检查上传重复失败: %w", err)
							} else {
								atomic.AddInt64(&initialize.StatsInfo.JumpCount, 1)
								logx.Logger.Info("跳过已上传", zap.String("Signature", v.Signature))
							}
							return nil
						}

						baseName := uploader.GenerateObjectName()
						objName := baseName + ".mp4"
						posterName := baseName + ".jpg"

						if err := uploader.UploadToTOS(gctx, advertiserID, objName, v.URL, posterName, v.PostUrl); err != nil {
							logx.Logger.Error("上传失败: ", zap.String("视频url", v.URL), zap.String("封面url", v.PostUrl), zap.Error(err))
							return fmt.Errorf("上传失败: %w", err)
						}

						//放入管道
						if sub := re.FindStringSubmatch(v.Filename); sub != nil {
							payload := Material{
								Name:      sub[1],
								File:      objName,
								Size:      v.Size,
								Dimension: "0",
								Duration:  v.Duration,
								UserID:    1,
								PackageID: pkgData.ID,
								Poster:    posterName,
								Signature: v.Signature,
							}
							//10缓冲的管道，第11个便会阻塞
							materialChan <- payload
						}

						return nil
					})
				}

				if err := g.Wait(); err != nil {
					logx.Logger.Error("部分素材上传tos任务执行失败", zap.Error(err))
				}

				time.Sleep(300 * time.Millisecond)
			}
		}
	}()

	var wg sync.WaitGroup
	var materialArr []Material

	//消费者
	for {
		select {
		case <-ctx.Done():
			logx.Logger.Info("任务取消，停止收集素材")
			return ctx.Err()
		case material, ok := <-materialChan:
			if !ok {
				logx.Logger.Info("素材通道已关闭，处理剩余素材", zap.Int("剩余条数", len(materialArr)))

				if len(materialArr) > 0 {
					//请求接口
					finalBat := make([]Material, len(materialArr))
					copy(finalBat, materialArr)
					wg.Add(1)

					go func(bat []Material) {
						defer wg.Done()
						addMaterial(ctx, bat)
					}(finalBat)
				}

				wg.Wait()

				logx.Logger.Info("广告主任务完成",
					zap.String("advertiser_id", advertiserID),
					zap.Int64("成功数", atomic.LoadInt64(&initialize.StatsInfo.Success)),
					zap.String("耗时", time.Since(start).String()),
				)

				return nil
			}
			materialArr = append(materialArr, material)
			//一直读到10个再执行中台记录处理
			if len(materialArr) >= 10 {
				bat := make([]Material, len(materialArr))
				copy(bat, materialArr)
				materialArr = materialArr[:0] //清空累积数组
				wg.Add(1)

				go func(bat []Material) {
					defer wg.Done()
					addMaterial(ctx, bat)
				}(bat)
			}
		}
	}
}

func addMaterial(ctx context.Context, b []Material) {
	//请求接口
	addMaterialURL := config.Cfg.AddMaterialURL
	body, _ := json.Marshal(b)
	// fmt.Println("打印一下数据", b)
	req2, err := http.NewRequestWithContext(ctx, http.MethodPost, addMaterialURL, bytes.NewReader(body))

	if err != nil {
		logx.Logger.Error("创建添加素材请求失败", zap.Error(err))
		return
	}

	req2.Header.Set("x-service-key", "gFnGn~58Hk^dF(+uEtc+cish")
	req2.Header.Set("Content-Type", "application/json")
	resp2, err := http.DefaultClient.Do(req2)

	if err != nil {
		logx.Logger.Error("调用添加素材接口失败", zap.Error(err))
		return
	}

	defer func() {
		if resp2 != nil && resp2.Body != nil {
			resp2.Body.Close()
		}
	}()

	respBody, _ := io.ReadAll(resp2.Body)

	if resp2.StatusCode != http.StatusOK {
		logx.Logger.Error("添加素材接口响应错误",
			zap.Int("status", resp2.StatusCode),
			zap.String("body", string(respBody)))
		return
	}

	//添加sqlite记录
	for _, s := range b {
		_, err = db.MarkUploaded(s.Signature)

		if err != nil {
			logx.Logger.Error("sqlite MarkUploaded出错", zap.Error(err))
			continue
		}

		atomic.AddInt64(&initialize.StatsInfo.Success, 1)
	}

	var addResp struct {
		Code int    `json:"code"`
		Msg  string `json:"msg"`
	}

	if err := json.Unmarshal(respBody, &addResp); err != nil {
		logx.Logger.Error("解析添加素材响应失败", zap.Error(err))
	}

	logx.Logger.Info("add-material 调用成功",
		zap.Int("code", addResp.Code),
		zap.String("msg", addResp.Msg))
}

func min(a, b int) int {
	if a < b {
		return a
	}

	return b
}
